/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */

package cgl.granules.dataset;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import cgl.granules.util.MarshallingUtil;
import cgl.narada.util.UUIDRetriever;

/**
 * This is an implementation of the <code>DataCollection</code> Interface.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class DatasetCollectionManagement implements DatasetCollection,
        DataAvailabilityObserver {
    private ConcurrentHashMap<String, Dataset> datasets;
    private ConcurrentHashMap<String, DataAvailabilityNotifier> availabilityNotifiers;
    private ConcurrentLinkedQueue<String> pausedNotifiers;
    private String identifier;
    private String moduleName = "DatasetCollectionManagement: ";

    private boolean closed = false;
    private boolean availabilityReported = false;
    private DatasetCollectionObserver collectionObserver;


    protected DatasetCollectionManagement() {
        initializeDataStructures();
        UUIDRetriever retriever = UUIDRetriever.getInstance();
        identifier = retriever.getRandomBasedUUIDAsString();
    }


    private void initializeDataStructures() {
        datasets = new ConcurrentHashMap<String, Dataset>();
        availabilityNotifiers =
            new ConcurrentHashMap<String, DataAvailabilityNotifier>();
        pausedNotifiers = new ConcurrentLinkedQueue<String>();
    }


    @Override
    public boolean isClosed() {
        return closed;
    }


    @Override
    public void initializeDatasets() throws DatasetException {
        Iterator<Dataset> iterator = getDatasets();

        while (iterator.hasNext()) {
            Dataset dataset = iterator.next();
            dataset.initializeDataset();
        }
    }


    @Override
    public int addDataset(Dataset dataset) throws DatasetException {
        String errorReport = null;
        if (dataset == null) {
            errorReport = moduleName + "The specified dataset is NULL!";
            throwDatasetException(errorReport);
        }
        String datasetIdentifier = dataset.getIdentifier();
        if (datasets.containsKey(datasetIdentifier)) {
            errorReport =
                moduleName + "Dataset (" + datasetIdentifier
                    + ") was previously added to the collection";
            throwDatasetException(errorReport);
        }
        datasets.put(datasetIdentifier, dataset);

        if (dataset.supportsDataAvailabilityNotifications()) {
            DataAvailabilityNotifier notifier =
                dataset.getDataAvailabilityNotifier();
            notifier.registerDataAvailabilityObserver(this);
            availabilityNotifiers.put(datasetIdentifier, notifier);
        }
        
        return datasets.size();
    }


    @Override
    public Dataset getDataset(String datasetIdentifier) throws DatasetException {
        checkDatasetIdentifier(datasetIdentifier);
        return datasets.get(datasetIdentifier);
    }


    @Override
    public Iterator<Dataset> getDatasets() {
        return datasets.values().iterator();
    }


    @Override
    public int getNumberOfDatasets() {
        return datasets.size();
    }


    @Override
    public boolean hasDataset() {
        if (datasets.isEmpty()) {
            return false;
        }
        return true;
    }


    @Override
    public boolean hasDataset(String datasetIdentifier) throws DatasetException {
        checkDatasetIdentifier(datasetIdentifier);
        if (datasets.containsKey(datasetIdentifier)) {
            return true;
        }
        return false;
    }


    @Override
    public int removeDataset(String datasetIdentifier) throws DatasetException {
        checkDatasetIdentifier(datasetIdentifier);
        if (!datasets.containsKey(datasetIdentifier)) {
            String errorReport =
                moduleName + "Trying to remove a non-existent dataset: "
                    + datasetIdentifier;
            throwDatasetException(errorReport);
        }
        datasets.remove(datasetIdentifier);
        availabilityNotifiers.remove(datasetIdentifier);

        return datasets.size();
    }


    /**
     * Check to see if the specified datasetIdentifier is NULL
     * 
     * @param datasetIdentifier
     * @throws DatasetException
     *             If the specified identifier is NULL
     */
    private void checkDatasetIdentifier(String datasetIdentifier)
        throws DatasetException {
        String errorReport = null;
        if (datasetIdentifier == null) {
            errorReport =
                moduleName + "The specified datasetIdentifier is NULL";
            throwDatasetException(errorReport);
        }
    }


    /**
     * Throws a data set exception based on the specified error report
     * 
     * @param errorReport
     *            The errorReport for the exception.
     * @throws DatasetException
     */
    private void throwDatasetException(String errorReport)
        throws DatasetException {
        DatasetException dse = new DatasetException(errorReport);
        throw dse;
    }


    protected DatasetCollectionManagement(byte[] marshalledBytes)
        throws IOException, DatasetException {
        MarshallingUtil marshaller = MarshallingUtil.getInstance();
        ByteArrayInputStream baInputStream =
            new ByteArrayInputStream(marshalledBytes);
        DataInputStream din = new DataInputStream(baInputStream);

        initializeDataStructures();
        DatasetFactory datasetFactory = DatasetFactory.getInstance();

        identifier = marshaller.readString(din);
        int numOfDatasets = din.readInt();
        for (int i = 0; i < numOfDatasets; i++) {
            int datasetLength = din.readInt();
            byte[] datasetBytes = new byte[datasetLength];
            din.readFully(datasetBytes);
            Dataset dataset = datasetFactory.getDataset(datasetBytes);
            dataset.initializeDataset();
            addDataset(dataset);
        }

    }


    @Override
    public byte[] getBytes() throws IOException {
        MarshallingUtil marshaller = MarshallingUtil.getInstance();
        byte[] marshalledBytes = null;
        ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
        DataOutputStream dout = new DataOutputStream(baOutputStream);

        marshaller.writeString(dout, identifier);
        int numOfDatasets = datasets.size();
        dout.writeInt(numOfDatasets);

        Iterator<Dataset> iterator = getDatasets();

        int written = 0;
        while (iterator.hasNext()) {
            Dataset dataset = iterator.next();
            byte[] datasetBytes = dataset.getBytes();
            dout.writeInt(datasetBytes.length);
            dout.write(datasetBytes);
            written++;
        }

        if (written != numOfDatasets) {
            throw new IOException(moduleName
                + "Collection modified during getBytes()");
        }

        dout.flush();
        marshalledBytes = baOutputStream.toByteArray();
        return marshalledBytes;
    }


    @Override
    public boolean isDataAvailable() {
        Iterator<Dataset> iterator = getDatasets();

        while (iterator.hasNext()) {
            Dataset dataset = iterator.next();
            if (dataset.isDataAvailable()) {
                return true;
            }
        }
        return false;
    }


    @Override
    public void setDatasetCollectionObserver(DatasetCollectionObserver observer) {
        collectionObserver = observer;
    }


    @Override
    public String getIdentifier() {
        return identifier;
    }


    @Override
    public synchronized void onDataAvailability(String datasetIdentifier) {
        //if (!availabilityReported) {
            collectionObserver.onDataAvailability();
         //   availabilityReported = true;
        //} else {
           // pauseAvailabilityNotifications(datasetIdentifier);
       // }
    }


    @Override
    public void onDatasetClosure(String datasetIdentifier) {
        try {
            int numOfDatasetsRemaining = removeDataset(datasetIdentifier);
            /**
             * Make sure that data availability notifications are resumed on
             * other datasets that are part of this collection
             */
            if (numOfDatasetsRemaining == 0) {
                System.out.println(moduleName
                    + "All underlying datasets have been closed");
                closed = true;
                collectionObserver.onDatasetCollectionClosure();
            }

        } catch (DatasetException se) {
            System.out.println(moduleName
                + "Problems processing dataset closure " + se.getMessage());
        }
    }


    private void pauseAvailabilityNotifications(String datasetIdentifier) {
        try {
            DataAvailabilityNotifier notifier =
                availabilityNotifiers.get(datasetIdentifier);
            if (notifier == null) {
                System.out.println(moduleName
                    + "Unable to locate DataAvailability Notifier for: "
                    + datasetIdentifier);
                return;
            }
            notifier.pauseAvailabilityNotifications(this);
            pausedNotifiers.add(datasetIdentifier);
        } catch (DatasetException dse) {
            String errorReport =
                "Problems pausing availability notifications on dataset "
                    + datasetIdentifier + ":" + dse.getMessage();
            System.out.println(errorReport);
        }
    }


    @Override
    public void resumeAvailabilityNotifications() {
        availabilityReported = false;
        if (pausedNotifiers.isEmpty()) {
            return;
        }

        while (!pausedNotifiers.isEmpty()) {
            String datasetIdentifier = pausedNotifiers.poll();
            if (datasetIdentifier == null) {
                return;
            }
            DataAvailabilityNotifier notifier =
                availabilityNotifiers.get(datasetIdentifier);
            if (notifier == null) {
                System.out.println(moduleName
                    + "Unable to locate DataAvailability Notifier for: "
                    + datasetIdentifier);
                return;
            }

            try {
                notifier.resumeAvailabilityNotifications(this);
            } catch (DatasetException dse) {
                String errorReport =
                    "Problems resuming availability notifications on dataset "
                        + datasetIdentifier + ":" + dse.getMessage();
                System.out.println(errorReport);
            }/* end try-catch */
        } /* end while() */
    }


    public String toString() {
        String stringRep =
            "Dataset Collection: Number of Datasets=" + datasets.size();

        Iterator<Dataset> iterator = getDatasets();

        while (iterator.hasNext()) {
            Dataset dataset = iterator.next();
            stringRep += ("\n" + dataset.toString());
        }
        return stringRep;
    }


    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub

    }

}
